package net.bwie.jtp.dws.log.funcion;

import net.bwie.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    @Override
    public void apply(String key, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {
        // TODO: 2025/5/20 key分组字段  window表示时间窗口可以获取开始时间和结束时间 input表示窗口的数据
        // TODO: 2025/5/20 窗口开始时间和结束时间
        String start = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String end = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
        // TODO: 2025/5/20 窗口中的计算结果
        PageViewBean  bean = input.iterator().next();
        long pvCount = bean.getPvCount();
        long pvDuringTime = bean.getPvDuringTime();
        long uvCount = bean.getUvCount();
        long sessionCount = bean.getSessionCount();
        // TODO: 2025/5/20 输出
        String output = start+","+end+","+key+","+pvCount
                +","+uvCount+","+pvDuringTime+","+sessionCount+","+ System.currentTimeMillis();
        out.collect(output);

    }
}
